"""
Case Type   : gs_preupgradechk工具
Case Name   : gs_preupgradechk工具A3项对传统主备异常场景检测
Create At   : 2025/11/17
Owner       : chen
Description :
    1./tmp可用空间小于等于50M
    2./home可用空间小于等于50M
    3.$GAUSSHOME可用空间小于等于1GB
Expect      :
    1.成功
    2.成功
    3.成功
History     :
"""

import unittest
import os
import re
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

LOG = Logger()
primary_sh = CommonSH('PrimaryDbUser')


class Tools_Gs_Preupgradechk_TestCase0010(unittest.TestCase):
    def setUp(self):
        LOG.info(f'----{os.path.basename(__file__)} start----')
        self.pri_node = Node('PrimaryDbUser')
        self.pri_root = Node('PrimaryRoot')
        self.com = Common()
        LOG.info("---SetUp：检查数据库状态是否正常---")
        status = primary_sh.get_db_cluster_status()
        LOG.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)

    def test_gs_preupgradechk_tools(self):
        text = '----step1: /tmp可用空间小于等于50M,检测A3项; expect: 成功----'
        LOG.info(text)
        tmp_cmd = r"df -B MB /tmp  | awk 'NR==2 {{print $4}}'"
        LOG.info(tmp_cmd)
        tmp_res = self.pri_node.sh(tmp_cmd).result()
        LOG.info(tmp_res)
        tmp_res = tmp_res.replace("MB", "")
        tmp_size = int(tmp_res) - 10
        cmd = f'fallocate -l {tmp_size}MB /tmp/disk_full_case0010.txt'
        LOG.info(cmd)
        res = self.pri_root.sh(cmd).result()
        LOG.info(res)
        cmd = f"gs_preupgradechk -i A3"
        LOG.info(cmd)
        result = self.pri_node.sh(f"source {macro.DB_ENV_PATH};"
                                    f"{cmd}").result()
        LOG.info(result)
        self.assertIn('Successfully checked disk usage status', result)
        pattern = r'\[([a-zA-Z0-9_-]+)\][^[]*?\s*\[([^\]]+)\]'
        matches = re.findall(pattern, result)
        kw = {}
        for instance, status in matches:
            kw[instance] = status
        node_num = self.com.get_node_num(self.pri_node)
        self.assertEqual(len(kw.keys()), node_num)
        hostname = self.pri_node.sh(f"hostname").result()
        try:
            self.assertEqual(kw[hostname], 'Error')
        except KeyError:
            LOG.info('节点hostname收集错误,主动构造错误断言，跳出方法')
            self.assertTrue(False)
        LOG.info('----清理磁盘----')
        cmd = 'rm -rf /tmp/disk_full_case0010.txt'
        LOG.info(cmd)
        self.pri_root.sh(cmd).result()

        text = '----step2: /home可用空间小于等于50M,检测A3项; expect: 成功----'
        LOG.info(text)
        home_cmd = r"df -B MB /home  | awk 'NR==2 {{print $4}}'"
        LOG.info(home_cmd)
        home_res = self.pri_node.sh(home_cmd).result()
        LOG.info(home_res)
        home_res = home_res.replace("MB", "")
        home_size = int(home_res) - 10
        cmd = f'fallocate -l {home_size}MB /home/disk_full_case0010.txt'
        LOG.info(cmd)
        res = self.pri_root.sh(cmd).result()
        LOG.info(res)
        cmd = f"gs_preupgradechk -i A3"
        LOG.info(cmd)
        result = self.pri_node.sh(f"source {macro.DB_ENV_PATH};"
                                    f"{cmd}").result()
        LOG.info(result)
        self.assertIn('Successfully checked disk usage status', result)
        matches = re.findall(pattern, result)
        kw = {}
        for instance, status in matches:
            kw[instance] = status
        self.assertEqual(len(kw.keys()), node_num)
        try:
            self.assertEqual(kw[hostname], 'Error')
        except KeyError:
            LOG.info('节点hostname收集错误,主动构造错误断言，跳出方法')
            self.assertTrue(False)
        LOG.info('----清理磁盘----')
        cmd = 'rm -rf /home/disk_full_case0010.txt'
        LOG.info(cmd)
        self.pri_root.sh(cmd).result()

        text = '----step3: $GAUSSHOME可用空间小于等于1GB; expect: 成功----'
        LOG.info(text)
        cmd = f'echo $GAUSSHOME'
        result = self.pri_node.sh(f"source {macro.DB_ENV_PATH};"
                                    f"{cmd}").result()
        LOG.info(result)
        res = self.get_first_directory_linux(result)
        gs_cmd = f"df -B MB {res}  | awk " \
            r"'NR==2 {{print $4}}'"
        LOG.info(gs_cmd)
        gs_res = self.pri_node.sh(gs_cmd).result()
        LOG.info(gs_res)
        gs_res = gs_res.replace("MB", "")
        gs_size = int(gs_res) - 100
        wr_file = f'{result}/disk_full_case0010.txt'
        cmd = f'fallocate -l {gs_size}MB {wr_file}'
        LOG.info(cmd)
        res = self.pri_root.sh(cmd).result()
        LOG.info(res)
        cmd = f"gs_preupgradechk -i A3"
        LOG.info(cmd)
        result = self.pri_node.sh(f"source {macro.DB_ENV_PATH};"
                                    f"{cmd}").result()
        LOG.info(result)
        self.assertIn('Successfully checked disk usage status', result)
        matches = re.findall(pattern, result)
        kw = {}
        for instance, status in matches:
            kw[instance] = status
        self.assertEqual(len(kw.keys()), node_num)
        try:
            self.assertEqual(kw[hostname], 'Error')
        except KeyError:
            LOG.info('节点hostname收集错误,主动构造错误断言，跳出方法')
            self.assertTrue(False)
        LOG.info('----清理磁盘----')
        cmd = f'rm -rf {wr_file}'
        LOG.info(cmd)
        self.pri_root.sh(cmd).result()
        
    def get_first_directory_linux(self, path):
        if not path or path == '/':
            return path
        path = path.rstrip('/')
        parts = [p for p in path.split('/') if p]
        if not parts:
            return '/'
        if path.startswith('/'):
            return '/' + parts[0]
        else:
            return parts[0] 

    def tearDown(self):
        LOG.info('--------清理环境--------')
        status = primary_sh.get_db_cluster_status("detail")
        LOG.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        LOG.info(f"------{os.path.basename(__file__)} end------")
